package cn.bytets.rpc.common.timer;

import cn.bytets.rpc.common.utils.NamedTheadFactory;

import java.util.Locale;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 单层哈希类型 时间轮实现
 */
public class HashWheelTimer implements Timer {

    /**
     * 存储时间轮的数组
     */
    private final HashedWheelBucket[] wheel;
    /**
     * 时间刻度执行间隔时间,默认100毫秒
     */
    private final long tickDuration;
    /**
     * 掩码,主要是为计算 存储时间轮的数组下标,以便能够环形应用
     * mask = wheel-1;
     */
    private final int mask;

    private final CountDownLatch startTimeInitialized = new CountDownLatch(1);

    private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();
    private final Queue<HashedWheelTimeout> cancelledTimeouts = new LinkedBlockingQueue<>();
    private final AtomicLong pendingTimeouts = new AtomicLong(0);

    private final AtomicIntegerFieldUpdater<HashWheelTimer> fieldUpdater = AtomicIntegerFieldUpdater.newUpdater(HashWheelTimer.class,"wheelState");
    /**
     * 状态 0 初始化,1 开始,2结束
     */
    public volatile int wheelState = 0;

    private final Worker worker = new Worker();

    private final Thread thread;

    private volatile long startTime;
    /**
     * 时间轮槽位 通过 tick&mask 计算
     */
    private volatile long tick;

    public HashWheelTimer(){
        this(100,TimeUnit.MILLISECONDS,
                new NamedTheadFactory("timer-task",true),512);
    }
    public HashWheelTimer(long tickDuration, TimeUnit unit,int ticksPerWheel){
        this(tickDuration,unit,new NamedTheadFactory("timer-task",true),ticksPerWheel);
    }
    public HashWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit){
        this(tickDuration,unit,threadFactory,512);
    }
    /**
     *
     * @param tickDuration 时间轮刻度间隔执行时间
     * @param unit
     * @param threadFactory
     * @param ticksPerWheel 存储时间轮数组长度
     */
    public HashWheelTimer(long tickDuration, TimeUnit unit, ThreadFactory threadFactory,int ticksPerWheel){
        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }
        if (tickDuration <= 0) {
            throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
        }
        if (ticksPerWheel <= 0) {
            throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
        }
        this.wheel = createWheel(ticksPerWheel);

        this.mask = wheel.length - 1;
        // 将刻度时间转换为纳米单位
        this.tickDuration = unit.toNanos(tickDuration);
        if(this.tickDuration >= Long.MAX_VALUE/wheel.length){
            throw new IllegalArgumentException(String.format(
                    "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
                    tickDuration, Long.MAX_VALUE / wheel.length));
        }
        this.thread = threadFactory.newThread(worker);
    }
    /**
     * @param delay 延迟执行时间数
     * @param unit  时间单位
     * @return
     */
    @Override
    public Timeout newTimeout(TimerTask task,long delay, TimeUnit unit) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }
        /**
         * 开始初始化线程
         */
        start();
        /**
         * 过期时间
         */
        long deadline = System.nanoTime()+unit.toNanos(delay)-startTime;
        if (delay > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }
        HashedWheelTimeout timeout = new HashedWheelTimeout(task,this,deadline);
        //timeouts.add(timeout);
        addWheelBucket(timeout);
        return timeout;
    }
    private void start(){
        switch (fieldUpdater.get(this)){
            case 0:
                if (fieldUpdater.compareAndSet(this, 0, 1)) {
                    thread.start();
                }
                break;
            case 1:
                break;
            case 2:
                throw new IllegalStateException("cannot be started once stopped");
            default:
                throw new Error("Invalid WorkerState");
        }
        /**
         * 等待线程执行
         */
        while (startTime==0L){
            try {
                startTimeInitialized.await();
            } catch (InterruptedException e) {
            }
        }
    }

    private final static class HashedWheelTimeout implements Timeout{

        /**
         * 链表指针
         */
        private HashedWheelTimeout next;
        private HashedWheelTimeout prev;

        /**
         * 添加超时的存储桶
         */
        private HashedWheelBucket bucket;
        /**
         * 当前任务
         */
        private TimerTask task;

        private Timer timer;
        /**
         * 过期时间(纳米数)
         */
        private final long deadline;

        private volatile boolean cancel = false;

        public HashedWheelTimeout(TimerTask task,Timer timer,long deadline){
            this.task  = task;
            this.timer = timer;
            this.deadline = deadline;
        }
        /**
         * 返回创建此句柄的 {@link Timer}。
         *
         * @return
         */
        @Override
        public Timer timer() {
            return this.timer;
        }

        /**
         * 返回时间轮执行的任务
         */
        @Override
        public TimerTask task() {
            return this.task;
        }

        /**
         * 判断此{@link TimerTask}是否过期
         */
        @Override
        public boolean isExpired() {
            return false;
        }

        /**
         * 判断此{@link TimerTask}是否取消
         */
        @Override
        public boolean isCancelled() {
            return cancel;
        }

        /**
         * 取消{@link TimerTask}
         *
         * @return 如果成功则返回ture, 否则false
         */
        @Override
        public boolean cancel() {
            cancel = true;
            return cancel;
        }

        public long getDeadline(){
            return this.deadline;
        }

        public void execute(){
            try {
                //TODO 这里要判断状态
                this.task.run(this);
            } catch (Exception e){

            }
        }
    }

    /**
     * 存储哈希时间轮的槽位,使用双向链表数据结构
     */
    private static final class HashedWheelBucket {
        /**
         * 链表指针
         */
        HashedWheelTimeout head;
        HashedWheelTimeout tail;

        void addTimeout(HashedWheelTimeout timeout){
            assert timeout.bucket == null;
            timeout.bucket = this;
            if(null==head){
                head = tail = timeout;
                return;
            }
            tail.next = timeout;
            timeout.prev = tail;
            tail = timeout;
        }

        /**
         * 处理任务
         * @param deadline
         */
        void expireTimeouts(long deadline){
            HashedWheelTimeout timeout = this.head;
            while (null!=timeout){
                HashedWheelTimeout next = timeout.next;
                if(timeout.isCancelled()){
                    next = remove(timeout);
                } else {
                    remove(timeout);
                    if (timeout.deadline <= deadline) {
                        timeout.execute();
                    } else {
                        // 放错插槽,这里不应该发生(例如:插槽只有128,当时超时时间设置过长导致在放插槽放错,因为插槽是循环应用的)
                        throw new IllegalStateException(String.format(
                                "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                    }
                }
                timeout = next;
            }
        }

        /**
         * 删除链表中的指针,使其置空方便GC回收
         * @param timeout
         * @return
         */
        public HashedWheelTimeout remove(HashedWheelTimeout timeout){
            HashedWheelTimeout next = timeout.next;
            if(null!=timeout.prev){
                timeout.prev.next = next;
            }
            if (null!=timeout.next){
                timeout.next.prev = timeout.prev;
            }
            if (head == timeout){
                if(tail == timeout){
                    head = null;
                    tail = null;
                } else {
                    head = next;
                }
            } else if(tail == timeout){
                tail = timeout.prev;
            }
            timeout.prev = null;
            timeout.next = null;
            timeout.bucket = null;
            return next;
        }
    }

    /**
     *
     * @param ticksPerWheel 存储时间轮数组长度(2的n次幂)
     * @return
     */
    private static HashedWheelBucket[] createWheel(int ticksPerWheel){
        if (ticksPerWheel <= 0) {
            throw new IllegalArgumentException(
                    "ticksPerWheel must be greater than 0: " + ticksPerWheel);
        }
        if (ticksPerWheel > 1073741824) {
            throw new IllegalArgumentException(
                    "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
        }
        ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
        HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
        for (int i=0;i<ticksPerWheel;i++){
            wheel[i] = new HashedWheelBucket();
        }
        return wheel;
    }

    /**
     * 计算过期时间存储到数组的哪个槽位
     * @param timeout
     */
    private void addWheelBucket(HashedWheelTimeout timeout){
        long tempTick = timeout.getDeadline()/tickDuration;
        final long ticks = Math.max(tempTick,tick);
        int stopIndex = (int) (ticks & mask);

        HashedWheelBucket bucket = wheel[stopIndex];
        bucket.addTimeout(timeout);
    }
    /**
     * 这里主要是为能将 数组的长度为 2的n次幂,方便计算下标
     * @param ticksPerWheel
     * @return
     */
    private static int normalizeTicksPerWheel(int ticksPerWheel) {
        int normalizedTicksPerWheel = ticksPerWheel - 1;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 1;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 2;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 4;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 8;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 16;
        return normalizedTicksPerWheel + 1;
    }

    /**
     * 处理延迟任务线程
     */
    private final class Worker implements Runnable{

        private long tick;

        @Override
        public void run() {
            /**
             * 时间轮的起始时间,后续的任务都已该时间为起点计算
             */
            startTime = System.nanoTime();
            if(startTime == 0L){
                startTime = 1L;
            }
            startTimeInitialized.countDown();
            while (1==fieldUpdater.get(HashWheelTimer.this)){
                long waitTime = waitForNextTick();
                if(waitTime>0){
                    int idx = (int)tick&mask;
                    HashedWheelBucket bucket = wheel[idx];
                    bucket.expireTimeouts(waitTime);
                    tick++;
                }
            }
        }

        /**
         * 等待下一个时间刻度
         * @return
         */
        public long waitForNextTick(){
            long delay = tickDuration*(tick+1);
            while (true){
                final long currentTime = System.nanoTime() - startTime;
                /**
                 * 计算时间轮下一刻度休眠时间(转为毫秒单位)
                 * 这里为什么是加 999999?
                 */
                long sleepTimeMs = (delay-currentTime+999999)/1000000;
                /**
                 * 表示时间轮已经是当前刻度了
                 */
                if(sleepTimeMs<=0){
                    if (currentTime == Long.MIN_VALUE) {
                        return -Long.MAX_VALUE;
                    } else {
                        return currentTime;
                    }
                }
                if (isWindows()) {
                    sleepTimeMs = sleepTimeMs / 10 * 10;
                }
                try {
                    /**
                     * 释放当前线程锁,休眠到下一个时间轮刻度
                     */
                    Thread.sleep(sleepTimeMs);
                } catch (InterruptedException e) {
                    if (fieldUpdater.get(HashWheelTimer.this) == 2) {
                        return Long.MIN_VALUE;
                    }
                }
            }
        }
    }
    private boolean isWindows() {
        return System.getProperty("os.name", "").toLowerCase(Locale.US).contains("win");
    }
}
